obsidian_curriculum_page

Amazon Bedrock (Claude 5) & 로컬 벡터 DB (FAISS) 연동 RAG 실습 가이드

이 페이지는 광주소프트웨어마이스터고등학교(GSM) AI 해커톤 교육의 실습 코드 가이드라인입니다. 아래 코드를 순서대로 복사하여 EC2 서버 내에서 실행하고 서비스를 구축할 수 있습니다.

1. 실습 환경 준비 (라이브러리 설치)

EC2 인스턴스에 접속한 후, 아래 명령어를 실행하여 필요한 패키지를 전역 환경에 설치합니다.

# 시스템 패키지 업데이트 및 AWS CLI 설치
sudo apt-get update
sudo apt-get install -y unzip curl
curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
unzip awscliv2.zip
sudo ./aws/install

# 파이썬 RAG 핵심 패키지 설치
pip3 install boto3 streamlit langchain langchain-aws langchain-community faiss-cpu pypdf --break-system-packages

💡 AWS 자격 증명 점검 명령어:

aws sts get-caller-identity

출력 결과에 EC2 인스턴스에 할당된 Role(예: gsm-student-role 또는 gsm-instructor-role)이 정상적으로 나타나는지 확인합니다.

2. 실습 소스 코드 파일 구성

Note

💡 리눅스 터미널(EC2)에서 파일 생성 및 코드 작성하는 방법 (Nano 편집기)

리눅스 터미널 환경에서 아래 소스 코드를 생성하고 붙여넣으려면 nano 편집기를 사용합니다:

  1. 파일 열기/생성: 터미널에 nano <파일명>을 입력합니다. (예: nano bedrock_simple_test.py)
  2. 코드 붙여넣기: 가이드의 소스코드를 전체 복사한 후, 터미널 창에 마우스 오른쪽 버튼을 클릭하여 붙여넣습니다. (Mac 터미널은 Cmd + V)
  3. 저장 및 종료:
    • 저장: 키보드 단축키 Ctrl + O ➡️ Enter 입력
    • 종료: 키보드 단축키 Ctrl + X 입력 (터미널 원래 화면으로 복귀)

① 패키지 버전 목록 (requirements.txt)

프로젝트 폴더 내에 생성하여 버전 충돌을 방지합니다.

boto3>=1.34.0
streamlit>=1.32.0
langchain>=0.2.0
langchain-aws>=0.1.0
langchain-community>=0.2.0
faiss-cpu>=1.7.4
pypdf>=4.0.0

② Bedrock 기초 호출 테스트 (bedrock_simple_test.py)

IAM 권한이 올바르게 잡혀 Bedrock API가 성공적으로 통신하는지 가장 먼저 점검하는 코드입니다.

import json
import boto3

def test_bedrock():
    # Bedrock Runtime Client 초기화 (자동으로 EC2 IAM 인스턴스 역할 인지)
    bedrock_runtime = boto3.client(
        service_name='bedrock-runtime',
        region_name='us-east-1'
    )
    
    # Claude 5 Sonnet 교차 리전 추론 프로필 ID
    model_id = "us.anthropic.claude-sonnet-5"
    
    prompt = "AWS EC2 인스턴스 프로파일을 통해 Amazon Bedrock Claude 5 연동에 최종 성공했습니다. 축하 메시지 한 줄 출력해줘."
    
    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 300,
        "messages": [
            {"role": "user", "content": prompt}
        ]
    })
    
    print("Sending prompt to Bedrock Claude...")
    try:
        response = bedrock_runtime.invoke_model(
            modelId=model_id,
            body=body
        )
        
        response_body = json.loads(response.get('body').read())
        answer = response_body['content'][0]['text']
        print("\n--- Claude Response ---")
        print(answer)
        print("-----------------------")
        
    except Exception as e:
        print(f"Error calling Bedrock: {e}")

if __name__ == "__main__":
    test_bedrock()

③ 로컬 벡터 DB 빌더 (bedrock_faiss_indexer.py)

S3 버킷에서 원천 지식 파일을 다운로드하여 텍스트 청킹을 수행한 뒤, 로컬 FAISS 벡터 데이터베이스(faiss_index/)로 생성하는 코드입니다.

import os
import boto3
from langchain_aws import BedrockEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter

def build_local_vector_db():
    # TODO: 본인 계정의 S3 버킷명으로 반드시 변경해 주세요!
    bucket_name = "gsm-instructor-bucket-tbit-498307943987"
    s3_file_key = "s3_test.txt"
    local_file_name = "downloaded_context.txt"
    
    # 1. S3에서 파일 다운로드
    print("1. Downloading raw file from S3...")
    s3 = boto3.client('s3')
    try:
        s3.download_file(bucket_name, s3_file_key, local_file_name)
        print(f"   Successfully downloaded '{s3_file_key}' to '{local_file_name}'")
    except Exception as e:
        print(f"   Error downloading from S3: {e}")
        print("   If downloading fails, checking for local 'rich_context.txt' as fallback...")
        if os.path.exists("rich_context.txt"):
            local_file_name = "rich_context.txt"
            print("   Using local 'rich_context.txt' fallback.")
        else:
            print("   No fallback file found. Indexing aborted.")
            return
    
    # 2. 파일 텍스트 로드 및 청킹 (Chunking)
    print("2. Splitting text into chunks...")
    with open(local_file_name, 'r', encoding='utf-8') as f:
        text = f.read()
        
    # 청크 크기 100자, 중복 20자 설정
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)
    docs = text_splitter.create_documents([text])
    print(f"   Created {len(docs)} text chunks.")
    
    # 3. Bedrock Titan Embedding v2 모델 설정 (버지니아 리전)
    print("3. Initializing Bedrock Titan Text Embeddings...")
    bedrock_client = boto3.client(service_name='bedrock-runtime', region_name='us-east-1')
    embeddings = BedrockEmbeddings(
        client=bedrock_client,
        model_id="amazon.titan-embed-text-v2:0"
    )
    
    # 4. 로컬 FAISS 벡터 DB 구축 및 로컬 저장
    print("4. Generating vectors and saving to local FAISS index...")
    db = FAISS.from_documents(docs, embeddings)
    db.save_local("faiss_index")
    print("   Success! Vector DB saved locally as folder './faiss_index'\n")

if __name__ == "__main__":
    build_local_vector_db()

④ Streamlit 기반 실시간 스트리밍 웹 RAG 챗봇 (app.py)

Day 3에 학생들이 최종적으로 완성할 웹 UI 애플리케이션 데모입니다. 로컬 FAISS 벡터 검색 결과를 병합하여 Claude 5 API를 호출하고 한 글자씩 타이핑되듯 보여주는 실시간 스트리밍을 지원하며, 추가 라우팅 페이지를 통해 웹 UI에서 직접 S3 문서 업로드 및 로컬 벡터 DB(FAISS) 즉시 갱신을 수행할 수 있습니다.

import os
import json
import boto3
import streamlit as st
from langchain_aws import BedrockEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1. Page Config
st.set_page_config(page_title="GSM AI Hackathon Portal", page_icon="🤖", layout="centered")

# AWS 및 S3/Bedrock 설정
# TODO: 본인 계정의 S3 버킷명으로 반드시 변경해 주세요!
BUCKET_NAME = "gsm-instructor-bucket-tbit-498307943987"
S3_FILE_KEY = "s3_test.txt"  # 기본 RAG 참고용 파일 키

# 2. AWS Client Initialization
@st.cache_resource
def get_aws_clients():
    s3_client = boto3.client('s3')
    bedrock_client = boto3.client(service_name='bedrock-runtime', region_name='us-east-1')
    return s3_client, bedrock_client

s3_client, bedrock_client = get_aws_clients()

# 3. Bedrock Embeddings Config
def get_embeddings():
    return BedrockEmbeddings(
        client=bedrock_client,
        model_id="amazon.titan-embed-text-v2:0"
    )

# 4. Local FAISS DB Loader (동적 갱신 반영을 위해 캐싱 제거)
def load_vector_db():
    embeddings = get_embeddings()
    try:
        db = FAISS.load_local("faiss_index", embeddings, allow_dangerous_deserialization=True)
        return db
    except Exception as e:
        return None

# 5. Multi-Page Navigation (Sidebar)
st.sidebar.title("📌 메뉴 구성")
page = st.sidebar.radio("원하시는 페이지를 선택하세요:", ["💬 RAG 챗봇", "📁 S3 문서 업로드 & DB 갱신"])

# ==========================================
# PAGE 1: RAG Chatbot
# ==========================================
if page == "💬 RAG 챗봇":
    st.title("💬 RAG Q&A 챗봇 서비스")
    st.write("로컬 벡터 데이터베이스(FAISS)를 조회하여 팩트 기반의 답변을 실시간 스트리밍합니다.")
    
    # 디스크에서 실시간으로 인덱스 로드
    db = load_vector_db()
    
    if db is None:
        st.warning("로컬 FAISS 인덱스를 불러올 수 없습니다. '📁 S3 문서 업로드 & DB 갱신' 메뉴에서 인덱스를 먼저 빌드해 주세요.")

    # Initialize chat history
    if "messages" not in st.session_state:
        st.session_state.messages = []

    # Display chat messages from history on app rerun
    for message in st.session_state.messages:
        with st.chat_message(message["role"]):
            st.markdown(message["content"])

    # Stream Response Generator from Bedrock
    def stream_bedrock_response(prompt_text):
        body = json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 800,
            "messages": [
                {"role": "user", "content": prompt_text}
            ]
        })
        try:
            response = bedrock_client.invoke_model_with_response_stream(
                modelId="us.anthropic.claude-sonnet-5",
                body=body
            )
            for event in response.get('body'):
                chunk = json.loads(event.get('chunk').get('bytes').decode('utf-8'))
                if chunk.get('type') == 'content_block_delta':
                    yield chunk.get('delta', {}).get('text', '')
        except Exception as e:
            yield f"\n\n⚠️ Error calling Bedrock: {e}"

    # Chat logic
    if db is not None:
        if query := st.chat_input("질문을 입력해 주세요..."):
            with st.chat_message("user"):
                st.markdown(query)
            st.session_state.messages.append({"role": "user", "content": query})
            
            # FAISS similarity search (k=4)
            docs = db.similarity_search(query, k=4)
            context_text = "\n---\n".join([doc.page_content for doc in docs])

            
            # Augmented prompt
            augmented_prompt = f"""
당신은 친절한 안내원입니다. 아래 주어진 [참고문서]를 바탕으로 질문에 답하세요.
참고문서에 없는 질문은 모른다고 명확히 답해야 합니다.

[참고문서]
{context_text}

질문: {query}
답변:
"""
            with st.chat_message("assistant"):
                response_placeholder = st.empty()
                full_response = ""
                for chunk in stream_bedrock_response(augmented_prompt):
                    full_response += chunk
                    response_placeholder.markdown(full_response + "▌")
                response_placeholder.markdown(full_response)
                
            st.session_state.messages.append({"role": "assistant", "content": full_response})

# ==========================================
# PAGE 2: Document Upload & Index Update
# ==========================================
elif page == "📁 S3 문서 업로드 & DB 갱신":
    st.title("📁 S3 문서 업로드 및 벡터 DB 갱신")
    st.write("로컬 PC의 문서를 S3 버킷으로 업로드하고, 실시간으로 벡터 데이터베이스(FAISS)를 갱신합니다.")
    
    st.info(f"현재 타겟 S3 버킷: `{BUCKET_NAME}`")
    
    # 1. File Uploader Form
    uploaded_file = st.file_uploader("업로드할 텍스트 파일(.txt)을 선택해 주세요", type=["txt"])
    
    if uploaded_file is not None:
        file_details = {"FileName": uploaded_file.name, "FileType": uploaded_file.type, "FileSize": f"{uploaded_file.size} Bytes"}
        st.write(file_details)
        
        # 원클릭 업로드 및 인덱스 갱신 통합 버튼
        if st.button("S3 업로드 및 벡터 DB 즉시 갱신 실행"):
            progress_bar = st.progress(0)
            status_text = st.empty()
            
            try:
                # 1단계: S3 업로드
                status_text.text("1/5 단계: S3 버킷으로 파일 업로드 중...")
                s3_client.upload_fileobj(uploaded_file, BUCKET_NAME, S3_FILE_KEY)
                progress_bar.progress(20)
                
                # 2단계: S3에서 최신 문서 다운로드 (동기화 검증)
                status_text.text("2/5 단계: S3에서 방금 업로드된 최신 문서 다운로드 중...")
                local_file_name = "downloaded_context.txt"
                s3_client.download_file(BUCKET_NAME, S3_FILE_KEY, local_file_name)
                progress_bar.progress(40)
                
                # 3단계: 텍스트 로드 및 청킹
                status_text.text("3/5 단계: 문서 데이터를 읽고 텍스트 청크 분할 중...")
                with open(local_file_name, 'r', encoding='utf-8') as f:
                    text = f.read()
                text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
                docs = text_splitter.create_documents([text])

                progress_bar.progress(60)
                
                # 4단계: 임베딩 및 인덱스 생성
                status_text.text("4/5 단계: Titan 임베딩 생성 및 벡터 DB 빌드 중...")
                embeddings = get_embeddings()
                db = FAISS.from_documents(docs, embeddings)
                progress_bar.progress(80)
                
                # 5단계: 로컬 저장
                status_text.text("5/5 단계: 로컬 디스크에 faiss_index 저장 완료 중...")
                db.save_local("faiss_index")
                progress_bar.progress(100)
                
                st.success(f"성공! 파일이 `s3://{BUCKET_NAME}/{S3_FILE_KEY}` 경로로 업로드되었으며, 벡터 DB 갱신이 최종 완료되었습니다. (총 {len(docs)}개의 청크 데이터베이스화 성공)")
                status_text.text("준비 완료! '💬 RAG 챗봇' 메뉴로 이동하여 질문해 보세요.")
                
                # 대화 기록 세션 초기화
                st.session_state.messages = []
                
            except Exception as e:
                st.error(f"작업 도중 에러가 발생했습니다: {e}")

⚡ 3. 정상 실행 검증 명령어

① 로컬 벡터 DB 빌드 실행

python3 bedrock_faiss_indexer.py

체크포인트: 실행 폴더 경로에 faiss_index/ 폴더가 생성되었고 안에 index.faiss, index.pkl 파일이 들어있는지 ls -la faiss_index/ 명령어로 확인합니다.

② Streamlit 웹 애플리케이션 가동

python3 -m streamlit run app.py --server.address 0.0.0.0

체크포인트: 서버 구동 후 본인 컴퓨터 웹 브라우저에서 http://<EC2-퍼블릭IP>:8501 주소로 정상 접속되는지 확인합니다.

💡 RAG 기능 자가진단 질문 (환각 현상 방지 테스트)

4. Streamlit 백그라운드 상시 구동 및 관리 (SSH 종료 후 유지)

학생들이 터미널 세션을 종료해도 웹 서버가 꺼지지 않고 계속 기동되도록 관리하는 두 가지 백그라운드 구동 기법입니다.

방법 A: nohup 사용 (단일 명령어로 간편 구동)

방법 B: screen 사용 (가상 다중 창 터미널 관리)